// com/reliable/service/listener/ChatMessageImportListener.java
package com.reliable.service.listener;

import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.read.listener.ReadListener;
import com.reliable.api.dto.excel.ChatMessageExcelDTO;
import com.reliable.common.enums.DataSourceType;
import com.reliable.framework.datasource.DynamicDataSourceContextHolder;
import com.reliable.service.excel.ChatMessageRow;
import com.reliable.service.service.ChatMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ChatMessageImportListener implements ReadListener<ChatMessageExcelDTO> {

    private static final Logger log = LoggerFactory.getLogger(ChatMessageImportListener.class);

    private final Long taskId;
    private final Long uploadId; // 👈 新增：上传记录ID（来自MySQL）
    private final ChatMessageService chatMessageService;

    private final List<ChatMessageRow> batch = new ArrayList<>();
    private final Set<String> uniqueSessionIds = new HashSet<>(); // 👈 新增：收集唯一会话ID

    private final AtomicInteger totalRows = new AtomicInteger(0);     // 总行数（非空行）
    private final AtomicInteger successRows = new AtomicInteger(0);   // 成功行数
    private final AtomicInteger failedRows = new AtomicInteger(0);    // 失败行数

    private LocalDateTime minTime = null;
    private LocalDateTime maxTime = null;

    // 👇 修改构造函数：增加 uploadId
    public ChatMessageImportListener(Long taskId, Long uploadId, ChatMessageService chatMessageService) {
        this.taskId = taskId;
        this.uploadId = uploadId;
        this.chatMessageService = chatMessageService;
    }

    @Override
    public void invoke(ChatMessageExcelDTO data, AnalysisContext context) {
        totalRows.incrementAndGet();

        // 跳过完全空的行
        if (data.getSessionId() == null && data.getRole() == null && data.getContent() == null) {
            failedRows.incrementAndGet(); // 穗空行也算失败
            return;
        }

        try {
            // 角色标准化
            String role = "client"; // 默认
            if ("bot".equalsIgnoreCase(data.getRole()) || "agent".equalsIgnoreCase(data.getRole())) {
                role = "agent";
            }

            // 解析时间
            LocalDateTime createdAt = null;
            try {
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                createdAt = LocalDateTime.parse(data.getCreatedAtStr().trim(), formatter);
            } catch (Exception e) {
                log.warn("⚠️ 时间解析失败，跳过行 {}: {}", context.readRowHolder().getRowIndex() + 1, data.getCreatedAtStr());
                failedRows.incrementAndGet();
                return;
            }

            // 更新时间范围
            if (minTime == null || createdAt.isBefore(minTime)) minTime = createdAt;
            if (maxTime == null || createdAt.isAfter(maxTime)) maxTime = createdAt;

            // 构建对象
            ChatMessageRow row = new ChatMessageRow(
                    data.getSessionId(),
                    role,
                    data.getContent(),
                    createdAt,
                    "excel",
                    taskId,
                    data.getCompanyId() != null ? Long.valueOf(data.getCompanyId()) : null,
                    data.getVisitorId(),
                    data.getAgentAccount(),
                    data.getVisitorIp(),
                    data.getReferrerUrl(),
                    data.getPageUrl()
            );

            batch.add(row);
            uniqueSessionIds.add(data.getSessionId()); // 👈 收集 session_id

            successRows.incrementAndGet();

            // 每 1000 行批量插入
            if (batch.size() >= 1000) {
                flushBatch();
            }

            int current = successRows.get() + failedRows.get();
            if (current % 10000 == 0) {
                log.info("已处理行数: {}", current);
            }

        } catch (Exception e) {
            log.error("行解析异常，行号: {}", context.readRowHolder().getRowIndex() + 1, e);
            failedRows.incrementAndGet();
        }
    }

    private void flushBatch() {
        if (!batch.isEmpty()) {
            try {
                chatMessageService.batchInsertChatMessages(batch);
                batch.clear();
            } catch (Exception e) {
                throw new RuntimeException("批量插入 chat_messages 失败", e);
            }
        }
    }

    @Override
    public void doAfterAllAnalysed(AnalysisContext context) {
        log.info("📊 [DEBUG] doAfterAllAnalysed 开始，uploadId={}", uploadId);
        log.info("📊 [DEBUG] 总行数: {}, 成功行数: {}, 失败行数: {}",
                totalRows.get(), successRows.get(), failedRows.get());
        log.info("📊 [DEBUG] uniqueSessionIds size: {}", uniqueSessionIds.size());

        flushBatch(); // 处理剩余数据

        // 确保后续操作在 PG 数据源下执行
        String originalDataSource = DynamicDataSourceContextHolder.getDataSourceType();
        try {
            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.PG.name());

            if (minTime != null && maxTime != null) {
                chatMessageService.createPartitionsIfNeeded(minTime, maxTime);
            }

            if (!uniqueSessionIds.isEmpty()) {
                try {
                    log.info("📤 准备调用 saveUploadSessionRef，uploadId={}, 会话数={}", uploadId, uniqueSessionIds.size());
                    chatMessageService.saveUploadSessionRef(uploadId, uniqueSessionIds);
                    log.info("✅ 保存 upload_session_ref 完成，uploadId={}, 会话数={}", uploadId, uniqueSessionIds.size());
                } catch (Exception e) {
                    log.error("❌ 保存 upload_session_ref 失败", e);
                    throw new RuntimeException("保存会话映射失败", e);
                }
            } else {
                log.warn("⚠️ uniqueSessionIds 为空，跳过保存 upload_session_ref");
            }

        } finally {
            // 恢复原始上下文（虽然异步线程通常不需要，但更安全）
            if (originalDataSource != null) {
                DynamicDataSourceContextHolder.setDataSourceType(originalDataSource);
            } else {
                DynamicDataSourceContextHolder.clearDataSourceType();
            }
        }

        log.info("✅ 导入完成，总行数: {}, 成功: {}, 失败: {}",
                totalRows.get(), successRows.get(), failedRows.get());
    }

    // 👇 提供统计信息给外部（用于更新 MySQL upload_record）
    public int getTotalRows() { return totalRows.get(); }
    public int getSuccessRows() { return successRows.get(); }
    public int getFailedRows() { return failedRows.get(); }
}